package com.hub.realtime.resource;


import java.util.List;
import javax.servlet.http.HttpServletResponse;

import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.hub.realtime.flinkshims.core.model.SqlError;
import com.hub.realtime.resource.domain.ResFlinkConf;
import com.hub.realtime.resource.dto.input.*;
import com.hub.realtime.resource.dto.output.UploadResult;
import com.hub.realtime.resource.service.IResFlinkConfService;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import com.hub.realtime.common.annotation.Log;
import com.hub.realtime.common.core.controller.BaseController;
import com.hub.realtime.common.core.domain.AjaxResult;
import com.hub.realtime.common.enums.BusinessType;
import com.hub.realtime.resource.domain.ResFlinkJob;
import com.hub.realtime.resource.service.IResFlinkJobService;
import com.hub.realtime.common.utils.poi.ExcelUtil;
import com.hub.realtime.common.core.page.TableDataInfo;
import org.springframework.web.multipart.MultipartFile;

/**
 * 作业管理中心Controller
 *
 * @author wupeng
 * @date 2022-01-18
 */
@RestController
@RequestMapping("/resource/job")
public class ResFlinkJobController extends BaseController {
    @Autowired
    private IResFlinkJobService resFlinkJobService;

    @Autowired
    IResFlinkConfService flinkConfService;

    /**
     * 查询作业管理中心列表
     */
    @PreAuthorize("@ss.hasPermi('resource:job:list')")
    @GetMapping("/list")
    public TableDataInfo list(FlinkJobQueryParam param) {
        Page<ResFlinkJob> page = resFlinkJobService.list(param);
        return getDataTable(page.getRecords(), page.getTotal());
    }

    /**
     * 导出作业管理中心列表
     */
    @PreAuthorize("@ss.hasPermi('resource:job:export')")
    @Log(title = "作业管理中心", businessType = BusinessType.EXPORT)
    @PostMapping("/export")
    public void export(HttpServletResponse response, FlinkJobQueryParam param) {
        List<ResFlinkJob> list = resFlinkJobService.list(param).getRecords();
        ExcelUtil<ResFlinkJob> util = new ExcelUtil<ResFlinkJob>(ResFlinkJob.class);
        util.exportExcel(response, list, "作业管理中心数据");
    }

    /**
     * 获取作业管理中心详细信息
     */
    @PreAuthorize("@ss.hasPermi('resource:job:query')")
    @GetMapping(value = "/{id}")
    public AjaxResult getInfo(@PathVariable("id") Long id) {
        return AjaxResult.success(resFlinkJobService.getBaseMapper().selectById(id));
    }

    /**
     * 新增作业管理中心
     */
    @PreAuthorize("@ss.hasPermi('resource:job:add')")
    @Log(title = "作业管理中心", businessType = BusinessType.INSERT)
    @PostMapping
    public AjaxResult saveInfo(@RequestBody ResFlinkConf resFlinkConf) {
        return toAjax(resFlinkJobService.saveInfo(resFlinkConf));
    }

    /**
     * 修改作业管理中心
     */
    @PreAuthorize("@ss.hasPermi('resource:job:edit')")
    @Log(title = "作业管理中心", businessType = BusinessType.UPDATE)
    @PutMapping
    public AjaxResult edit(@RequestBody ResFlinkConf resFlinkJob) {
        return toAjax(resFlinkJobService.saveInfo(resFlinkJob));
    }

    /**
     * 删除作业管理中心
     */
    @PreAuthorize("@ss.hasPermi('resource:job:remove')")
    @Log(title = "作业管理中心", businessType = BusinessType.DELETE)
    @DeleteMapping("/{ids}")
    public AjaxResult remove(@PathVariable Long[] ids) {
        resFlinkJobService.deleteInfos(ids);
        return AjaxResult.success("删除成功！");
    }

    @PostMapping(value = "/createUploadFile")
    public AjaxResult createUploadFile(@RequestPart("file") MultipartFile[] file,
                                       @RequestParam("applicationId") Long applicationId) {
        List<UploadResult> list = resFlinkJobService.createUploadFile(file, applicationId);
        return AjaxResult.success(list);
    }

    @PostMapping(value = "/deleteHdfsTmp")
    public AjaxResult deleteHdfsTmp(@RequestBody List<DepParam> params) {
        resFlinkJobService.deleteHdfsTmp(params);
        return AjaxResult.success("删除成功！");
    }

    @GetMapping(value = "/getFlinkConfByJobId/{jobId}")
    public AjaxResult getFlinkConfByJobId(@PathVariable("jobId") Long jobId) {
        ResFlinkConf flinkConf = flinkConfService.getFlinkConfByJobId(jobId);
        return AjaxResult.success(flinkConf);
    }

    @PutMapping(value = "/deployejob/{jobId}")
    public AjaxResult deployejob(@PathVariable("jobId") Long jobId) {
        resFlinkJobService.deployejob(jobId);
        return AjaxResult.success("发布成功");
    }


    /**
     * 启动任务
     *
     * @return
     */
    @PostMapping(value = "/startJob")
    public AjaxResult startJob(@RequestBody JobStartAndStopParam param) {
        resFlinkJobService.startJob(param);
        return AjaxResult.success("启动成功");
    }

    /**
     * 停止作业
     *
     * @param param
     * @return
     */
    @PostMapping(value = "/stopJob")
    public AjaxResult stopJob(@RequestBody StopTriggerSavePoint param) {
        resFlinkJobService.stopJob(param);
        return AjaxResult.success("停止成功");
    }


    /**
     * 校验sql是否合法
     *
     * @param sqlQueryParam
     * @return
     */
    @PostMapping(value = "/verifySql")
    public AjaxResult verifySql(@RequestBody SqlQueryParam sqlQueryParam) {
        SqlError sqlError = resFlinkJobService.verifySql(sqlQueryParam);
        if (sqlError != null) {
            return AjaxResult.success(sqlError.getException());
        } else {
            return AjaxResult.success(null);
        }
    }

    @GetMapping(value = "/getErrorLogByJobId/{jobId}")
    public AjaxResult getErrorLogByJobId(@PathVariable("jobId") Long jobId) {
        return AjaxResult.success(resFlinkJobService.getErrorLogByJobId(jobId));
    }


}

